package cn.fan.mqtt.server;

import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.Channel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

/**
 * 对MQTT客户端发送消息后，处理的返回消息，基于MQTT协议的，需要MQTT协议的主要内容
 */
@Component
public class BootMqttMsgBack {


    private static Logger log = LoggerFactory.getLogger(BootMqttMsgBack.class);
    private static ConcurrentHashMap channelMap = new ConcurrentHashMap();
    private static ConcurrentHashMap<String, Map<String, Channel>> topicMap = new ConcurrentHashMap<String, Map<String, Channel>>();
    private static ConcurrentHashMap<String, Map<String, String>> device2Topic = new ConcurrentHashMap<>();
    private static ConcurrentHashMap<Integer,MqttMessage>   qos2Message=new ConcurrentHashMap<Integer, MqttMessage>();
    protected static AttributeKey<Boolean> _login = AttributeKey.valueOf("login");
    protected static AttributeKey<String> _deviceId = AttributeKey.valueOf("deviceId");
    private static String _userName = "colin";
    private static String _passWord = "85719223";
    protected static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);

    /**
     * 确认连接请求
     *
     * @param channel
     * @param
     */
    public static void connack(Channel channel, MqttConnectReturnCode connectReturnCode) {
        //	构建返回报文， 可变报头
        MqttConnAckVariableHeader mqttConnAckVariableHeaderBack = new MqttConnAckVariableHeader(connectReturnCode, true);
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0x02);
        //	构建CONNACK消息体
        MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeaderBack, mqttConnAckVariableHeaderBack);
        log.info("back--" + connAck.toString());
        channel.writeAndFlush(connAck);
    }

    /**
     * 根据qos发布确认
     *
     * @param channel
     * @param mqttMessage
     */
    public static void puback(Channel channel, MqttMessage mqttMessage) {
        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttPublishMessage.fixedHeader();
        int messageId = mqttPublishMessage.variableHeader().packetId();
        MqttQoS qos = (MqttQoS) mqttFixedHeaderInfo.qosLevel();

        byte[] headBytes = new byte[mqttPublishMessage.payload().readableBytes()];
        mqttPublishMessage.payload().readBytes(headBytes);
        String data = new String(headBytes);
        System.out.println("publish data--" + data);
        executorService.execute(() -> {
            switch (qos) {
                case AT_MOST_ONCE:        //	至多一次
                    break;
                case AT_LEAST_ONCE:        //	至少一次
                    //	构建返回报文， 可变报头
                    MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                    //	构建返回报文， 固定报头
                    MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBACK, mqttFixedHeaderInfo.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeaderInfo.isRetain(), 0x02);
                    //	构建PUBACK消息体
                    MqttPubAckMessage pubAck = new MqttPubAckMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
                    log.info("back--" + pubAck.toString());
                    channel.writeAndFlush(pubAck);
                    break;
                case EXACTLY_ONCE:        //	刚好一次
                    //	构建返回报文， 固定报头
                    MqttFixedHeader mqttFixedHeaderBack2 = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_LEAST_ONCE, false, 0x02);
                    //	构建返回报文， 可变报头
                    MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack2 = MqttMessageIdVariableHeader.from(mqttPublishMessage.variableHeader().packetId());
                    MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack2, mqttMessageIdVariableHeaderBack2);
                    log.info("back--" + mqttMessageBack.toString());
                    channel.writeAndFlush(mqttMessageBack);
                    break;
                default:
                    break;
            }
            MqttPublishMessage sendMessge = buildMessage(mqttPublishMessage.variableHeader().topicName(), headBytes, messageId);
//          fixme  如果有保留位消息，需要直接保存 何时发送机制还不清楚
            if (qos.value() < 2){
                sendQos0Msg(sendMessge);
            }else {
                qos2Message.put(messageId,sendMessge);
            }

        });


    }


    public static MqttPublishMessage buildMessage(String topic, byte[] byteBuf, int messageId) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttPublishVariableHeader mqttPublishVariableHeader = new MqttPublishVariableHeader(topic, messageId);
//           Unpooled.wrappedBuffer() 多次发送回
        return new MqttPublishMessage(mqttFixedHeader, mqttPublishVariableHeader, Unpooled.copiedBuffer(byteBuf));


    }

    private static void sendQos0Msg(MqttPublishMessage mqttPublishMessage) {
        String topic = mqttPublishMessage.variableHeader().topicName();
        Map<String, Channel> stringChannelMap = topicMap.get(topic);
        Collection<Channel> channels = stringChannelMap.values();
        channels.parallelStream().forEach(p->{
            p.writeAndFlush(mqttPublishMessage);
        });
//        for (Map.Entry<String, Channel> entry : stringChannelMap.entrySet()) {
//            System.out.println("key= " + entry.getKey() + " and value= " + entry.getValue());
//
//            entry.getValue().writeAndFlush(mqttPublishMessage);
//        }
    }
    public static void sendQos3Msg( MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        int messageId = messageIdVariableHeader.messageId();
        MqttPublishMessage publishMessage = (MqttPublishMessage )qos2Message.get(messageId);
        sendQos0Msg(publishMessage);
    }
    /**
     * 发布完成 qos2
     *
     * @param channel
     * @param mqttMessage
     */
    public static void pubcomp(Channel channel, MqttMessage mqttMessage) {
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文， 固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_MOST_ONCE, false, 0x02);
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader mqttMessageIdVariableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        MqttMessage mqttMessageBack = new MqttMessage(mqttFixedHeaderBack, mqttMessageIdVariableHeaderBack);
        log.info("back--" + mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }

    /**
     * 订阅确认
     *
     * @param channel
     * @param mqttMessage
     */
    public static void suback(Channel channel, MqttMessage mqttMessage) {
        MqttSubscribeMessage mqttSubscribeMessage = (MqttSubscribeMessage) mqttMessage;
        MqttMessageIdVariableHeader messageIdVariableHeader = mqttSubscribeMessage.variableHeader();
        //	构建返回报文， 可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        Set<String> topics = mqttSubscribeMessage.payload().topicSubscriptions().stream().map(mqttTopicSubscription -> mqttTopicSubscription.topicName()).collect(Collectors.toSet());

        List<Integer> grantedQoSLevels = new ArrayList<>(topics.size());
        for (int i = 0; i < topics.size(); i++) {
            String topicName = mqttSubscribeMessage.payload().topicSubscriptions().get(0).topicName();
            Map<String, Channel> stringObjectMap = topicMap.get(topicName);

            String deviceId = channel.attr(_deviceId).get();
            Map<String, String> topic2s = device2Topic.get(deviceId);
            if (topic2s == null) topic2s = new HashMap<>();
            if (stringObjectMap == null) stringObjectMap = new HashMap<>();
            stringObjectMap.put(deviceId, channel);
            topicMap.put(topicName, stringObjectMap);
            topic2s.put(topicName, "");
            device2Topic.put(deviceId, topic2s);
            grantedQoSLevels.add(mqttSubscribeMessage.payload().topicSubscriptions().get(i).qualityOfService().value());
        }
        //	构建返回报文	有效负载
        MqttSubAckPayload payloadBack = new MqttSubAckPayload(grantedQoSLevels);
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2 + topics.size());
        //	构建返回报文	订阅确认
        MqttSubAckMessage subAck = new MqttSubAckMessage(mqttFixedHeaderBack, variableHeaderBack, payloadBack);
        log.info("back--" + subAck.toString());
        channel.writeAndFlush(subAck);
    }

    /**
     * 取消订阅确认
     *
     * @param channel
     * @param mqttMessage
     */
    public static void unsuback(Channel channel, MqttMessage mqttMessage) {
        MqttUnsubscribeMessage mqttMessage1 = (MqttUnsubscribeMessage) mqttMessage;
        String topic = mqttMessage1.payload().topics().get(0);
        deleteTopic(channel, topic);
        MqttMessageIdVariableHeader messageIdVariableHeader = (MqttMessageIdVariableHeader) mqttMessage.variableHeader();
        //	构建返回报文	可变报头
        MqttMessageIdVariableHeader variableHeaderBack = MqttMessageIdVariableHeader.from(messageIdVariableHeader.messageId());
        //	构建返回报文	固定报头
        MqttFixedHeader mqttFixedHeaderBack = new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 2);
        //	构建返回报文	取消订阅确认
        MqttUnsubAckMessage unSubAck = new MqttUnsubAckMessage(mqttFixedHeaderBack, variableHeaderBack);
        log.info("back--" + unSubAck.toString());
        channel.writeAndFlush(unSubAck);
    }

    /**
     * 心跳响应
     *
     * @param channel
     * @param mqttMessage
     */
    public static void pingresp(Channel channel, MqttMessage mqttMessage) {
        //	心跳响应报文	11010000 00000000  固定报文
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessage mqttMessageBack = new MqttMessage(fixedHeader);
        log.info("back--" + mqttMessageBack.toString());
        channel.writeAndFlush(mqttMessageBack);
    }

    //==========================================================以下方法不是协议中的方法===========
    public static boolean login(Channel channel, MqttMessage mqttMessage) throws UnsupportedEncodingException {
        MqttConnectMessage mqttConnectMessage = (MqttConnectMessage) mqttMessage;
        MqttFixedHeader mqttFixedHeaderInfo = mqttConnectMessage.fixedHeader();
        MqttConnectVariableHeader mqttConnectVariableHeaderInfo = mqttConnectMessage.variableHeader();
        String deviceId = mqttConnectMessage.payload().clientIdentifier();
//        登录成功
        MqttConnectPayload payload = mqttConnectMessage.payload();
        String userName = payload.userName();

        String passWord = new String((payload.passwordInBytes() == null ? new byte[]{} : payload.passwordInBytes()), "UTF-8");

        if (StringUtils.isEmpty(deviceId)) {
            MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED;
            connack(channel, connectReturnCode);
            return false;
        }
        if (!(mqttConnectMessage.variableHeader().hasPassword() && mqttConnectMessage.variableHeader().hasUserName())) {
            MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED;
            connack(channel, connectReturnCode);
            return false;
        }
        if (!(_userName.equals(userName) && _passWord.equals(passWord))) {
            MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD;
            connack(channel, connectReturnCode);
            return false;
        }

        channel.attr(_login).set(true);
        channel.attr(_deviceId).set(deviceId);
        channelMap.put(deviceId, channel);
        connack(channel, MqttConnectReturnCode.CONNECTION_ACCEPTED);
        return true;

    }

    public static void disconnect(Channel channel) {
        String deviceId = channel.attr(_deviceId).get();
//        清除登录
        channelMap.remove(deviceId);
        Map<String, String> relation = device2Topic.get(deviceId);
//清除订阅
        if (relation != null) {
            for (Map.Entry<String, String> entry : relation.entrySet()) {
                String topicName = entry.getKey();
                if (!StringUtils.isEmpty(topicName)) deleteTopic(channel, topicName);
            }
        }
//        清除订阅设备关联缓存
        device2Topic.remove(deviceId);
        channel.close();
    }

    public static void deleteTopic(Channel channel, String topicName) {
        Map<String, Channel> stringChannelMap = topicMap.get(topicName);
        stringChannelMap.remove(channel.attr(_deviceId).get());
    }
}